package com.gopay.scheduler.shedulerx.event;

import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import com.gopay.scheduler.shedulerx.util.ThreadPoolUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;

/**
 * @author 19371
 * 可改造成MQ形式
 */
public class JobEventBus {

    private final static Logger LOGGER = LoggerFactory.getLogger(JobEventBus.class);

    private final EventBus eventBus;

    private static volatile boolean isRegistered = false;

    public static volatile JobEventBus jobEventBus;

    private static final ThreadPoolExecutor busPool = new ThreadPoolExecutor(1, 10, 5, TimeUnit.MINUTES, new LinkedBlockingDeque<Runnable>(1000),
            new ThreadFactory() {
                int index = 0;

                @Override
                public Thread newThread(Runnable runnable) {
                    ++this.index;
                    return new Thread(runnable, "gopay-event-thread-" + this.index);
                }
            }, new RejectedExecutionHandler() {
            private int policyCount = 0;
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                LOGGER.error(String.format("the gopay scheduler event queue is full, the concurrent queue size = {%s};execute the rejected policy; rejectedCount={%s}",busPool.getQueue().size(),++policyCount));
                if (!executor.isShutdown()) {
                    r.run();
                }
            }
    });

    private JobEventBus(EventBus eventBus) {
        this.eventBus = eventBus;
        register();
    }

    public static JobEventBus getSingle(){
        if(jobEventBus == null){
            synchronized (JobEventBus.class){
                if(jobEventBus == null){
                   jobEventBus =  new JobEventBus(new AsyncEventBus(busPool));
                }
            }
        }
        return jobEventBus;
    }

    private void register() {
        eventBus.register(JobRdbListener.getSingle());
        isRegistered = true;
    }


    public void post(final JobEvent event) {
        if (isRegistered) {
            eventBus.post(event);
        }
    }

    public void shutdown(){
        String poolName = String.format("from class {%s} , and poolName is {%s}",this.getClass().getName(),"事件线程池");
        ThreadPoolUtil.shutdownGracefully(poolName,busPool,TimeUnit.MINUTES,5);
    }

}
